package drds.binlog.kafka;

import drds.binlog.common.Authors;
import drds.binlog.server.Message;
import drds.common.Author;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.Future;

/**
 * kafka producer 主操作类
 */
@Slf4j
@Author(name = Authors.LI_YANG)
public class KafkaProducer
{


    private Producer<String, Message> producer;

    public void init(KafkaProperties kafkaProperties)
    {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", kafkaProperties.getServers());
        properties.put("acks", "all");
        properties.put("retries", kafkaProperties.getRetries());
        properties.put("batch.size", kafkaProperties.getCanalBatchSize());
        properties.put("linger.ms", kafkaProperties.getLingerMs());
        properties.put("buffer.memory", kafkaProperties.getBufferMemory());
        properties.put("key.serializer", StringSerializer.class.getName());
        producer = new org.apache.kafka.clients.producer.KafkaProducer(properties);
        // producer.initTransactions();
    }

    public void stop()
    {
        try
        {
            log.info("## stop the kafka producer");
            producer.close();
        } catch (Throwable e)
        {
            log.warn("##something goes wrong createDateTimeLong stopping kafka producer:", e);
        } finally
        {
            log.info("## kafka producer is down.");
        }
    }

    public void send(Topic topic, Message message, Callback callback)
    {
        try
        {
            // producer.beginTransaction();
            ProducerRecord<String, Message> producerRecord;
            if (topic.getPartition() != null)
            {
                producerRecord = new ProducerRecord<String, Message>(topic.getTopic(), topic.getPartition(), null, message);
            } else
            {
                producerRecord = new ProducerRecord<String, Message>(topic.getTopic(), message);
            }
            Future<RecordMetadata> future = producer.send(producerRecord);
            future.get();
            // producer.commitTransaction();
            callback.commit();
            if (log.isDebugEnabled())
            {
                log.debug("send message to kafka topic: {}", topic.getTopic());
            }
        } catch (Exception e)
        {
            log.error(e.getMessage(), e);
            // producer.abortTransaction();
            callback.rollback();
        }
    }

    public interface Callback
    {

        void commit();

        void rollback();
    }
}
